package org.example.ItemCF.src.hdfs;

import java.io.IOException;
import java.net.URI;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FSDataInputStream;
import org.apache.hadoop.fs.FSDataOutputStream;
import org.apache.hadoop.fs.FileStatus;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.IOUtils;
import org.apache.hadoop.mapred.JobConf;

/**
 * HdfsDAO类用于操作Hadoop分布式文件系统（HDFS）。
 * 提供了包括创建目录、删除文件、列举目录内容、复制文件等方法。
 */
public class HdfsDAO {

    private static final String HDFS = "hdfs://192.168.134.132:9000/";

    /**
     * 构造函数，初始化HDFS路径和配置。
     *
     * @param conf 配置信息，用于HDFS操作。
     */
    public HdfsDAO(Configuration conf) {
        this(HDFS, conf);
    }

    /**
     * 构造函数，初始化HDFS路径和配置。
     *
     * @param hdfs HDFS的路径。
     * @param conf 配置信息，用于HDFS操作。
     */
    public HdfsDAO(String hdfs, Configuration conf) {
        this.hdfsPath = hdfs;
        this.conf = conf;
    }

    /**
     * HDFS文件路径
     */
    private String hdfsPath;

    /**
     * Hadoop配置信息
     */
    private Configuration conf;



    /**
     * 主函数，用于演示类中方法的使用。
     * 这个函数首先配置Hadoop作业的配置信息，然后创建一个HdfsDAO对象以用于后续的HDFS操作。
     * 具体操作包括将指定文件从一个目录复制到另一个目录，并列出目标目录下的文件列表。
     *
     * @param args 命令行参数。本函数中未使用该参数。
     * @throws IOException 读写文件时可能抛出的异常。
     */
    public static void main(String[] args) throws IOException {
        // 配置Hadoop作业的配置信息
        JobConf conf = config();
        // 创建HdfsDAO对象用于执行HDFS操作
        HdfsDAO hdfs = new HdfsDAO(conf);
        // 从源路径复制文件到目标路径
        hdfs.copyFile("src/main/java/org/example/ItemCF/data/small.csv", "/tmp/new");
        // 列出目标路径下的文件和目录
        hdfs.ls("/tmp/new");
    }


    /**
     * 配置Hadoop环境的函数。
     * 该函数不接受任何参数。
     *
     * @return JobConf对象，配置好了的Hadoop作业配置。
     */
    public static JobConf config() {
        // 创建一个新的JobConf对象，基于HdfsDAO类的配置
        JobConf conf = new JobConf(HdfsDAO.class);

        // 设置作业名称为"HdfsDAO"
        conf.setJobName("HdfsDAO");

        // 加载Hadoop核心配置、HDFS配置以及MapReduce配置
        conf.addResource("classpath:/hadoop/core-site.xml");
        conf.addResource("classpath:/hadoop/hdfs-site.xml");
        conf.addResource("classpath:/hadoop/mapred-site.xml");

        // 返回配置好的JobConf对象
        return conf;
    }


    /**
     * 在HDFS（Hadoop分布式文件系统）上创建目录。
     * 如果目录不存在，则会递归创建该目录及其所有父目录。
     *
     * @param folder 要在HDFS上创建的目录的路径。
     * @throws IOException 当读写文件时发生错误，会抛出此异常。
     */
    public void mkdirs(String folder) throws IOException {
        // 将传入的目录路径转换为Path对象
        Path path = new Path(folder);
        // 获取HDFS文件系统的实例
        FileSystem fs = FileSystem.get(URI.create(hdfsPath), conf);
        // 检查目录是否存在，如果不存在则创建
        if (!fs.exists(path)) {
            fs.mkdirs(path);
            System.out.println("Create: " + folder); // 打印创建目录的路径
        }
        // 关闭文件系统实例，释放资源
        fs.close();
    }


    /**
     * 递归删除HDFS上的目录及其内容。
     *
     * @param folder 要删除的目录路径，为HDFS上的绝对路径。
     * @throws IOException 在读写文件过程中可能抛出的异常。
     */
    public void rmr(String folder) throws IOException {
        // 将字符串路径转换为Path对象
        Path path = new Path(folder);
        // 获取HDFS文件系统的实例
        FileSystem fs = FileSystem.get(URI.create(hdfsPath), conf);
        // 设置删除文件时自动删除其所有内容且不可恢复
        fs.deleteOnExit(path);
        // 打印删除操作的信息
        System.out.println("Delete: " + folder);
        // 关闭文件系统实例，释放资源
        fs.close();
    }


    /**
     * 列出HDFS目录下的文件和子目录。
     *
     * @param folder 要列举内容的目录路径，是一个HDFS上的路径字符串。
     * @throws IOException 在读写文件时可能会抛出的异常。
     */
    public void ls(String folder) throws IOException {
        // 将传入的目录路径转换为Path对象
        Path path = new Path(folder);
        // 获取HDFS文件系统的实例
        FileSystem fs = FileSystem.get(URI.create(hdfsPath), conf);
        // 列出指定路径下的所有文件和目录
        FileStatus[] list = fs.listStatus(path);
        // 打印目录路径和分隔线
        System.out.println("ls: " + folder);
        System.out.println("==========================================================");
        // 遍历列出的文件和目录，打印每个的名称、是否为目录以及大小
        for (FileStatus f : list) {
            System.out.printf("name: %s, folder: %s, size: %d\n", f.getPath(), f.isDir(), f.getLen());
        }
        // 打印分隔线
        System.out.println("==========================================================");
        // 关闭FileSystem实例，释放资源
        fs.close();
    }


    /**
     * 在HDFS上创建新文件并写入内容。
     *
     * @param file 要创建的文件路径。此路径为HDFS上的路径。
     * @param content 要写入到新文件中的内容。
     * @throws IOException 当读写文件发生错误时抛出。
     */
    public void createFile(String file, String content) throws IOException {
        // 获取HDFS文件系统的实例
        FileSystem fs = FileSystem.get(URI.create(hdfsPath), conf);
        // 将内容转换为字节数组
        byte[] buff = content.getBytes();
        FSDataOutputStream os = null;
        try {
            // 在HDFS上创建新文件
            os = fs.create(new Path(file));
            // 将内容写入文件
            os.write(buff, 0, buff.length);
            System.out.println("Create: " + file);
        } finally {
            // 确保输出流被关闭
            if (os != null)
                os.close();
        }
        // 确保文件系统连接被关闭
        fs.close();
    }


    /**
     * 从本地文件系统复制文件到HDFS。
     *
     * @param local 本地文件路径。指定要从本地文件系统复制的文件的路径。
     * @param remote HDFS文件路径。指定文件复制到HDFS的目标路径。
     * @throws IOException 读写文件时可能抛出的异常。如果在复制过程中发生IO错误，则会抛出此异常。
     */
    public void copyFile(String local, String remote) throws IOException {
        // 获取HDFS文件系统的实例。
        FileSystem fs = FileSystem.get(URI.create(hdfsPath), conf);
        // 从本地文件系统复制文件到HDFS。
        fs.copyFromLocalFile(new Path(local), new Path(remote));
        // 打印复制操作的反馈信息。
        System.out.println("copy from: " + local + " to " + remote);
        // 关闭HDFS文件系统实例，释放资源。
        fs.close();
    }


    /**
     * 从HDFS下载文件到本地文件系统。
     *
     * @param remote HDFS文件路径。
     * @param local 本地文件路径。
     * @throws IOException 读写文件时可能抛出的异常。
     */
    public void download(String remote, String local) throws IOException {
        // 创建一个Path对象用于表示HDFS上的文件路径
        Path path = new Path(remote);
        // 获取一个文件系统实例，用于执行HDFS操作
        FileSystem fs = FileSystem.get(URI.create(hdfsPath), conf);
        // 从HDFS上将文件复制到本地文件系统
        fs.copyToLocalFile(path, new Path(local));
        // 打印下载信息
        System.out.println("download: from" + remote + " to " + local);
        // 关闭文件系统实例，释放资源
        fs.close();
    }


    /**
     * 在控制台输出HDFS文件的内容。
     *
     * @param remoteFile 要输出内容的HDFS文件路径。
     * @throws IOException 读写文件时可能抛出的异常。
     */
    public void cat(String remoteFile) throws IOException {
        // 将传入的文件路径转换为Path对象
        Path path = new Path(remoteFile);
        // 获取HDFS文件系统的实例
        FileSystem fs = FileSystem.get(URI.create(hdfsPath), conf);
        FSDataInputStream fsdis = null;
        // 打印正在处理的文件路径信息
        System.out.println("cat: " + remoteFile);
        try {
            // 打开文件并读取内容
            fsdis = fs.open(path);
            // 将文件内容复制到控制台输出流
            IOUtils.copyBytes(fsdis, System.out, 4096, false);
        } finally {
            // 确保输入流和文件系统连接被关闭
            IOUtils.closeStream(fsdis);
            fs.close();
        }
    }


}
